import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.Map;
import java.util.Queue;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.LinkedBlockingQueue;

/** A basic node in a network of computers. It has th ability to send and recieve
* messages.
*/
public class Node implements Runnable
{
    /** integer percentage of how often a node will fail-stop */
    public static final int FAIL_STOP_PCT = 50;
    
    /** integer percentage of how often a node will fail-restart */
    public static final int FAIL_RESTART_PCT = 0;
    
    protected static Map<Integer,Node> globalNodes = new HashMap<Integer,Node>();
    
    /** incoming data channel */
	LinkedBlockingQueue<String> commChannel;
    
    /** incoming message processing thread */
	Thread listener;
    NodeReceive listenerRcv;
    
    /** outgoing communication channels */
	Map<Integer, LinkedBlockingQueue<String>> commChannels;
    
    /** ID number */
	Integer nodeId;
	
    boolean incomingMessage = false;
    
    /** known information related to underlying computation. All information */
	Set<String> information; // all information
    
    /** number of messages that originate at this node */
    int localOrigin = 0;
    
    /** A count of how many copies of messages are recieved */
    long copyCount = 0;
    
    /** Thread used as a flag to signal end of a run */
    Thread isRunning = null;
    
    /** coverage, computed at death or competion */
    double coverage = 0;
    
    /** Create a new Node with the given ID. All contaniers are initialized
    * and start the incoming message listener.
    */
	public Node(Integer nodeId)
	{
		this.nodeId = nodeId;
		commChannel = new LinkedBlockingQueue<String>();
		commChannels = new HashMap<Integer, LinkedBlockingQueue<String>>();
		information = new HashSet<String>();
		listener = new Thread(listenerRcv = new NodeReceive(this));
		listener.start();
        globalNodes.put(nodeId,this);
	}
	
    /** Create a new Node with the given ID and a connection to the given node.
    * All containers are initialized to reflect this and the incoming message
    * listener is started.
    */
	public Node(Integer nodeId, Node node)
	{
        this(nodeId);
        fullConnect(node);
		/*this.nodeId = nodeId;
		commChannel = new LinkedBlockingQueue<String>();
		commChannels = new HashMap<Integer, LinkedBlockingQueue<String>>();
		information = new HashSet<String>();
		commChannels.put(nodeId, node.commChannel);
		node.commChannels.put(nodeId, commChannel);
		listener = new Thread(listenerRcv = new NodeReceive(this));
		listener.start();*/
	}
	
    public long countCopies()
    {
        return copyCount;
    }
    
    public int countRecieves()
    {
        return information.size();
    }
    
    public int countOrigin()
    {
        return localOrigin;
    }
    
    /** send a message.
    * Node implementation is broadcast.
    */
	public void send(String message)
	{
        // broadcast to all neighbors
		for(LinkedBlockingQueue<String> q : commChannels.values())
		{
			q.add(message);
		}
	}
    
    /** Explicitly connect to another node in one direction.
    * this matches the channel out of 'this' to the input of
    * the given node n.
    */
    public void halfConnect(Node n)
    {
        // log n as a neighbor
        if(!commChannels.containsKey(n.nodeId))
            commChannels.put(n.nodeId, n.commChannel);
    }
    
    /** Explicitly connect to another node in both directions.
    */
    public void fullConnect(Node n)
    {
        // we connect to them
        halfConnect(n);
        
        // they connect to us
        n.halfConnect(this);
    }
	
    /** Process a given message appropriately.
    * e.g. forward a message, discard, store, etc.
    * The basic Node will broadcast any new message
    * and discard any previously heard message.
    */
    public void processMessage(String message)
    {
        if(!information.contains(message))
        {
            information.add(new String(message));
            send(message);
        }else
            copyCount++;
    }
    
    public String toString()
    {
        return "" + nodeId;
    }
    
    public void step()
    {
    }
    
    public void stopRunning()
    {
        isRunning = null;
        // no interrupt here because the
        // node only blocks to sleep
    }
    
    /** Send random messages at up to a 5 second interval.
    */
	public void run()
	{
		Random rand = new Random();
		//Queue<String> info = new LinkedList<String>();
		boolean done = false;
        isRunning = Thread.currentThread();
		while(isRunning != null)
		{
            // send a random message
            String newMsg = ((Integer)rand.nextInt(Integer.MAX_VALUE)).toString()+","+nodeId;
            information.add(newMsg);
            localOrigin++;
            send(newMsg);
            
			//System.out.println(information.size()+","+nodeId);
            
            // random wait to continue
			try
			{
				Thread.sleep(rand.nextInt(5000));
                
                // check for stop and restart fail
                int randFail = rand.nextInt(1000);
                if(nodeId == 0)
                {
                	
                }
                else if(nodeId < 11 && randFail - FAIL_STOP_PCT/5 < 0)
                {
                	System.out.println("Node "+nodeId+" failed!");
                	break;
                }
                else if(nodeId >= 11 && randFail - FAIL_STOP_PCT < 0)
                {
                	System.out.println("Node "+nodeId+" failed!");
                	break;
                }
                
                /*if(nodeId != 0 && randFail - FAIL_RESTART_PCT < 0)
                {
                    Thread.sleep(5500);
                    // do something else
                }*/
			}
			catch (InterruptedException e)
			{
				e.printStackTrace();
			}
            
            // do any needed actions at
            // a break in the computation
            step();
		}
        
        // see how covered you were
        die();
        
        listenerRcv.stopRunning(); // close listener, finished using
	}
    
    private void die()
    {
        long mSys = 0; // unique messages in the system
        
        for(Node n : globalNodes.values())
            mSys += n.countOrigin();
        
        coverage = (double)information.size()/mSys;
    }
    
    public double getCoverage()
    {
        return coverage;
    }
}
